package org.trade.goods.mq.processor;

import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.trade.common.constant.MQEnums;
import org.trade.common.protocol.goods.AddGoodsNumberReq;
import org.trade.common.protocol.mq.CancelOrderMQ;
import org.trade.common.rocketmq.IMessageProcessor;
import org.trade.entity.TradeMqConsumerLog;
import org.trade.entity.TradeMqConsumerLogExample;
import org.trade.entity.TradeMqConsumerLogKey;
import org.trade.goods.service.IGoodsService;
import org.trade.mapper.TradeMqConsumerLogMapper;

import com.alibaba.fastjson.JSON;

/*
 * 商品消息去重，保证幂等性。
 */
public class CancelOrderProcessor implements IMessageProcessor {
	public static final Logger LOGGER = LoggerFactory.getLogger(CancelOrderProcessor.class);
	@Autowired
	private IGoodsService goodsService;

	/* 消费日志表（trade-mq-consumer-log） */
	@Autowired
	private TradeMqConsumerLogMapper tradeMqConsumerLogMapper;

	public boolean handleMessage(MessageExt messageExt) {
		TradeMqConsumerLog mqConsumerLog = null;
		try {
			String groupName = "goods_orderTopic_cancel_group";
			String body = new String(messageExt.getBody(), "UTF-8");
			String msgId = messageExt.getMsgId();
			String tags = messageExt.getTags();
			String keys = messageExt.getKeys();
			LOGGER.info("user CancelOrderProcessor receive message:" + messageExt);
			TradeMqConsumerLogKey key = new TradeMqConsumerLogKey();
			// 联合主键groupName以及tag，
			key.setGroupName(groupName);
			key.setMsgKey(keys);
			key.setMsgTag(tags);
			// 从MQ，消费日志表
			mqConsumerLog = this.tradeMqConsumerLogMapper.selectByPrimaryKey(key);

			if (mqConsumerLog != null) {
				String consumerStatus = mqConsumerLog.getConsumerStatus();
				// 消息已经处理过了
				if (MQEnums.ConsumerStatusEnum.SUCCESS.getStatusCode().equals(consumerStatus)) {
					LOGGER.warn("已经处理过了，不能再处理了");
					return true;
				} else if (MQEnums.ConsumerStatusEnum.PROCESSING.getStatusCode().equals(consumerStatus)) {
					LOGGER.warn("正在处理稍后再试");
					return false;
				} else if (MQEnums.ConsumerStatusEnum.FAIL.getStatusCode().equals(consumerStatus)) {
					if (mqConsumerLog.getConsumerTimes() >= 3) {
						// 让这个消息不再重试，转人工处理
						LOGGER.warn("超过三次，不再处理");
						return true;
					}
					TradeMqConsumerLog updateMqConsumerLog = new TradeMqConsumerLog();
					updateMqConsumerLog.setGroupName(groupName);
					updateMqConsumerLog.setMsgKey(keys);
					updateMqConsumerLog.setMsgTag(tags);
					updateMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.PROCESSING.getStatusCode());
					TradeMqConsumerLogExample tradeMqConsumerLogExample = new TradeMqConsumerLogExample();
					tradeMqConsumerLogExample.createCriteria().andGroupNameEqualTo(mqConsumerLog.getGroupName())
							.andMsgKeyEqualTo(mqConsumerLog.getMsgKey()).andMsgTagEqualTo(mqConsumerLog.getMsgTag())
							.andConsumerTimesEqualTo(mqConsumerLog.getConsumerTimes());
					// 乐观锁的方式防止并发更新，只有一个处理者能成功
					int record = this.tradeMqConsumerLogMapper.updateByExampleSelective(updateMqConsumerLog,
							tradeMqConsumerLogExample);
					if (record <= 0) {
						LOGGER.warn("并发更新处理状态，一会儿重试");
						return false;
					}

				}
			} else {
				// 并发时用主键冲突控制
				try {
					TradeMqConsumerLog tradeMqConsumerLog = new TradeMqConsumerLog();
					tradeMqConsumerLog.setGroupName(groupName);
					tradeMqConsumerLog.setMsgKey(keys);
					tradeMqConsumerLog.setMsgTag(tags);
					tradeMqConsumerLog.setConsumerTimes(0);
					tradeMqConsumerLog.setMsgId(msgId);
					tradeMqConsumerLog.setMsgBody(body);
					tradeMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.PROCESSING.getStatusCode());
					this.tradeMqConsumerLogMapper.insertSelective(tradeMqConsumerLog);
				} catch (Exception e) {
					LOGGER.warn("主键冲突，说明有订阅者正在处理，稍后再试");
					return false;
				}
			}
			CancelOrderMQ cancelOrderMQ = JSON.parseObject(body, CancelOrderMQ.class);
			AddGoodsNumberReq addGoodsNumberReq = new AddGoodsNumberReq();
			addGoodsNumberReq.setGoodsId(cancelOrderMQ.getGoodsId());
			addGoodsNumberReq.setGoodsNumber(cancelOrderMQ.getGoodsNumber());
			addGoodsNumberReq.setOrderId(cancelOrderMQ.getOrderId());
			goodsService.addGoodsNumber(addGoodsNumberReq);

			// 更新处理成功
			TradeMqConsumerLog updateMqConsumerLog = new TradeMqConsumerLog();
			updateMqConsumerLog.setGroupName(groupName);
			updateMqConsumerLog.setMsgKey(keys);
			updateMqConsumerLog.setMsgTag(tags);
			updateMqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes() + 1);
			updateMqConsumerLog.setMsgId(msgId);
			updateMqConsumerLog.setMsgBody(body);
			updateMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.SUCCESS.getStatusCode());
			this.tradeMqConsumerLogMapper.updateByPrimaryKeySelective(updateMqConsumerLog);
			return true;
		} catch (Exception e) {
			TradeMqConsumerLog updateMqConsumerLog = new TradeMqConsumerLog();
			updateMqConsumerLog.setGroupName(mqConsumerLog.getGroupName());
			updateMqConsumerLog.setMsgKey(mqConsumerLog.getMsgKey());
			updateMqConsumerLog.setMsgTag(mqConsumerLog.getMsgTag());
			updateMqConsumerLog.setConsumerTimes(mqConsumerLog.getConsumerTimes() + 1);
			updateMqConsumerLog.setConsumerStatus(MQEnums.ConsumerStatusEnum.SUCCESS.getStatusCode());
			this.tradeMqConsumerLogMapper.updateByPrimaryKeySelective(updateMqConsumerLog);
			return false;
		}

	}

}
